<?php
/**
 * Created by PhpStorm.
 * User: BlueIdeaLirui
 * Date: 2019/11/8
 * Time: 18:29
 */

namespace console\controllers\bigdata;

use business\base\caches\DataCache;
use business\bigdata\services\queues\MaterialConsumerService;
use business\bigdata\services\queues\MaterialProducerService;
use common\queue\QueueConstants;

/**
 * 材料数据清洗
 * Class MaterialController
 * @package console\controllers\material
 */
class MaterialController extends ReceiveController
{
    /**
     * 材料生产者调度
     * php yii bigdata/material/producer
     * docker exec -i opensearch_php php /var/www/html/opensearch/protected/yii bigdata/material/producer
     * @param $action
     */
    public function actionProducer($action = 'full')
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $materialProducerService = new MaterialProducerService();
            if (is_callable([$materialProducerService, $action])) {
                call_user_func_array([$materialProducerService, $action], []);
            }
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 消费端实现
     * php yii bigdata/material/consume 'bigdata-material-index-queue' true
     * docker exec -i opensearch_php php /var/www/html/opensearch/protected/yii bigdata/material/consume 'bigdata-material-index-queue' true
     * @param $message
     * @param $messageId
     */
    public function actionProcess($message, $messageId)
    {
        try {
            $cacheKey = $message;
            $module   = 'material';
            if (!empty($cacheKey)) {
                $data = (new DataCache())->getRecordByHash(
                    $module . '-' . QueueConstants::QUEUE_CHACHE_NAME, $cacheKey);

                !empty($data)
                    && (new MaterialConsumerService())->process($data['values'], $cacheKey, $messageId);
            }
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 重发机制
     */
    public function actionResend()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            (new MaterialProducerService())->resend();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 清除已经消费过的队列数据
     */
    public function actionClearConsumerQueue()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            (new MaterialProducerService())->clearConsumerQueue();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 接收消息并调用业务处理
     * @param array $messageBody
     */
    protected function successed(\iQueue\MessageBody $messageBody)
    {
        // 获取消息内容
        $message = $messageBody->getMessage();
        $messageId = $messageBody->getMessageId();

        $jsonMsg = (is_array($message) || is_object($message)) ?  json_encode($message) : $message;

        // 执行脚本
        $cmd = sprintf("php %s %s '%s' %s", $this->getYii(), $this->queueAction, $jsonMsg, $messageId);
        $this->debug && print "正在执行:{$cmd}\n";
        $output = system($cmd);
        $this->debug && print "{$output}\n";
    }
}
